package com.xzy.rabbit.demo.delayqueue.b;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.xzy.rabbit.common.util.RabbitUtil;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 延迟队列——消息TTL
 *
 * @author xzy
 * @date 2021/9/2914:09
 * RabbitMQ可以对消息和队列设置TTL,消息的过期时间以两者之间TTL较小的那个数值为准
 */
class Main {

    /**
     * 发送的消息总数
     */
    private static final Integer MESSAGE_COUNT = 3;
    /**
     * 发送每条消息的延迟时间
     */
    private static final long WORK_TIME = 10000L;

    /**
     * 消息过期时间
     */
    private static final long MSG_TTL_1 = 1000L;
    private static final long MSG_TTL_2 = 3000L;

    /**
     * 交换机名称
     */
    private static final String EXCHANGE_NAME_DELAY = "exchange.delay.msg-ttl.delay";
    private static final String EXCHANGE_NAME_DEAD = "exchange.delay.msg-tll.dead";

    /**
     * 队列名称
     */
    private static final String QUEUE_NAME_DELAY = "queue.delay.msg-ttl.delay";
    private static final String QUEUE_NAME_DEAD = "queue.delay.mgs-ttl.dead";

    /**
     * 路由键
     */
    private static final String ROUTING_KEY_DELAY = "delay";
    private static final String ROUTING_KEY_DEAD = "msg-ttl";

    /*
     * Producer--------[延迟交换机]----k1+TTL----[延迟队列]----k2----[死信交换机]----k2----[死信队列]----Consumer
     */

    /**
     * 准备执行环境：交换机、队列、绑定
     */
    private static void prepare(Connection connection) throws IOException {

        // 1.获取信道
        Channel channel = connection.createChannel();

        // 2.声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME_DELAY, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(EXCHANGE_NAME_DEAD, BuiltinExchangeType.DIRECT);

        // 3.声明队列
        Map<String, Object> arguments = new HashMap<>(2);
        arguments.put("x-dead-letter-exchange", EXCHANGE_NAME_DEAD);
        arguments.put("x-dead-letter-routing-key", ROUTING_KEY_DEAD);
        channel.queueDeclare(QUEUE_NAME_DELAY, false, false, false, arguments);

        channel.queueDeclare(QUEUE_NAME_DEAD, false, false, false, null);

        // 4.绑定
        channel.queueBind(QUEUE_NAME_DELAY, EXCHANGE_NAME_DELAY, ROUTING_KEY_DELAY);
        channel.queueBind(QUEUE_NAME_DEAD, EXCHANGE_NAME_DEAD, ROUTING_KEY_DEAD);
    }

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1.获取连接
        Connection connection = RabbitUtil.getConnection();

        // 2.准备执行环境
        prepare(connection);

        // 3.发送、接收消息
        Thread producer1s = new Thread(new Producer(connection, EXCHANGE_NAME_DELAY, ROUTING_KEY_DELAY, MESSAGE_COUNT, WORK_TIME, MSG_TTL_1), "Producer-1s");
        Thread producer3s = new Thread(new Producer(connection, EXCHANGE_NAME_DELAY, ROUTING_KEY_DELAY, MESSAGE_COUNT, WORK_TIME, MSG_TTL_2), "Producer-3s");
        Thread consumer = new Thread(new ConsumerDead(connection, QUEUE_NAME_DEAD), "Consumer-dead");

        producer3s.start();
        producer1s.start();
        consumer.start();

        producer1s.join();
        producer3s.join();
        consumer.join();

        /*
         * 缺点：RabbitMQ只会检查队列中的第一条消息是否过期，如果过期则丢到死信队列。如果第一条信息的过期时间很长，
         * 第二条消息的过期时间很短，只有处理完第一条消息后才会处理第二条消息，此时第二条消息实际已经过期很久了。
         */
    }
}
